package WindowAndWatermark

import Source.SensorReading
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

class WatermarkTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //设置并行度
    env.setParallelism(1)
    //事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //设置周期
    env.getConfig.setAutoWatermarkInterval(150l)

    val inputStream = env.socketTextStream("localhost", 7777)

    //转换成样例类
    val dataStream = inputStream
      .map(data => {
        val arr = data
          .split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }).assignTimestampsAndWatermarks(
      new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.milliseconds(3)) {
        override def extractTimestamp(t: SensorReading) = t.timeStamp
      })

    //每十五秒统计窗口内个传感器的温度最小值,以及最新的时间戳
    val resultStream = dataStream
      .map(data => (data.id, data.temperature, data.timeStamp))
      .keyBy(_._1) //按照二元组第一个元素id分组
      .timeWindow(Time.seconds(5))
      .allowedLateness(Time.minutes(1)) //设置迟到数据
      .sideOutputLateData(new OutputTag[(String, Double, Long)]("late")) //迟到的流
      .reduce((CurRes, newData) => (CurRes._1, CurRes._2.min(newData._2), newData._3))

    resultStream.print("res")

    env.execute()
  }
}
